package rpc.core.plugin.impl;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;
import rpc.bootstrap.spring.BeanPostProcessor;
import rpc.core.FlowMap;
import rpc.core.plugin.ServicePlugin;
import rpc.core.plugin.impl.proxy.RpcClientServiceProxy;
import rpc.core.plugin.impl.proxy.ServiceProxy;
import rpc.utils.ReflectionUtils;

/*********
 *
 *
 *@author 197
 *@date 2020/8/5 12:36
 *
 **/


public class RpcRefererServicePlugin implements ServicePlugin {


  private static final Logger logger = LoggerFactory.getLogger(RpcRefererServicePlugin.class);
  private final Map<String, ServiceProxy> flowInfoMap = new HashMap<>();
  private final ApplicationContext context;
  private final Environment environment;
  private final BeanPostProcessor beanPostProcessor;
  private int maxThreads = 4;
  private int threads = 1;
  private ThreadPoolExecutor executor = null;

  public RpcRefererServicePlugin(ApplicationContext context, Environment environment, BeanPostProcessor beanPostProcessor) {
    this.context = context;
    this.environment = environment;
    this.beanPostProcessor = beanPostProcessor;
  }

  @Override
  public void init() {

    List<? extends Class<?>> refersList = this.beanPostProcessor.getRefersList();
    if (refersList.size() == 0) {
      logger.info("refersList.size 0");
      return;
    }
    threads = intConfig(environment, "rpc.client.threads", 8);
    maxThreads = intConfig(environment, "rpc.client.maxthreads", 8);
    int queueSize = intConfig(environment, "rpc.client.queue", 100000);
    refersList.forEach(clazz -> {
      String clazzName = clazz.getSimpleName().toLowerCase();
      int serviceId = ReflectionUtils.getFliedIntValue(clazz, "serviceId");
      Arrays.stream(clazz.getDeclaredMethods()).forEach(m -> {
        RpcClientServiceProxy serviceProxy = new RpcClientServiceProxy(m, context.getBean(clazz));
        flowInfoMap.put(clazzName + "." + m.getName().toLowerCase(), serviceProxy);
        flowInfoMap.put(serviceId + "." + ReflectionUtils.getFliedIntValue(clazz, m.getName() + "_MsgId"), serviceProxy);
      });
    });
    if (logger.isDebugEnabled()) {
      logger.debug("flowInfoMap: {}", String.join(",", flowInfoMap.keySet()));
    }
    ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("pool_rpc_client_%d").build();
    executor = new ThreadPoolExecutor(threads, maxThreads, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueSize), threadFactory);
    executor.prestartCoreThread();
  }

  @Override
  public Set<String> getFlowName() {
    return flowInfoMap.keySet();
  }

  @Override
  public void close() {
    if (Objects.nonNull(executor)) {
      executor.shutdown();
    }
  }

  @Override
  public Map<String, ServiceProxy> getFlowServiceProxy() {
    return this.flowInfoMap;
  }

  @Override
  public Map<String, Object> dump() {
    FlowMap flowMap = new FlowMap();
    flowMap.append("threads", threads);
    flowMap.append("maxThreads", maxThreads);
    return flowMap.toMap();
  }
}
